package CompletableFuture;

import com.sun.org.apache.bcel.internal.generic.RETURN;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Collectors;

@Slf4j
public class CompletableFutureTest {

    public static final String RETURN_VALUE = "我是返回值";

    @Test
    public void test01() throws InterruptedException, ExecutionException {
        CompletableFuture<String> res = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
            return "执行完毕";
        });

        log.info(res.get());
    }

    @Test
    public void test02() throws InterruptedException, ExecutionException {
        CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "执行完毕").whenCompleteAsync((t, e) -> {
            if (e == null) {
                log.info(t + "_成功拿到結果");
            }
        }).exceptionally(e -> {
            log.error("执行失败", e);
            return "執行失敗";
        });
        log.info(cf.get());
    }

    @Test
    public void test03() throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> 1).thenApply(r -> r * 2).thenApply(r -> r * 3).exceptionally(e -> {
            e.printStackTrace();
            return 0;
        });
        System.out.println(String.valueOf(cf.get()) );

    }

    @Test
    public void test04() throws InterruptedException {
        CompletableFuture.supplyAsync(() -> "執行完畢").thenAccept(res -> log.info(res));
        //可以拿到异常
        CompletableFuture.supplyAsync(() -> "執行完畢").whenCompleteAsync((res, e) -> log.info(res));
        Thread.sleep(1000);
    }

    @Test
    public void test05() {
        CompletableFuture.supplyAsync(() -> "1").thenApply(s -> "2").whenComplete((r, e) -> {
            if (e == null) {
                log.info("執行完畢:" + r);
            } else {
                log.info("出现异常[{}]", e.getMessage());
            }
        });
    }

    @Test
    public void test06() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> log.info("无返回值得"));
        cf.get();
    }

    @Test
    public void test07() throws ExecutionException, InterruptedException {
        log.info(CompletableFuture.supplyAsync(() -> RETURN_VALUE).exceptionally(e -> "出现异常").get());
    }

    @Test
    public void test08() {
        //thenAccept 消费者 只消耗结果没有返回
        CompletableFuture.supplyAsync(() -> RETURN_VALUE).thenAccept(res -> log.info("执行成功，返回值：" + res));
    }

    @Test
    public void test09() {
        //thenApply fn类型 消耗完需要返回值 也就是需要可以一直无限调用
        CompletableFuture.supplyAsync(() -> RETURN_VALUE).thenApply(res -> res + " 增加处理逻辑1").thenApply(res -> res + " 增加处理逻辑2 ").thenAccept(log::info);
    }

    @Test
    public void test10() {
        //异常处理
        CompletableFuture.supplyAsync(() -> 1 / 0).exceptionally((e) -> {
            log.error("error", e);
            return 0;
        }).thenAccept(r -> log.info(String.valueOf(r)));
    }

    @Test
    public void test11() throws ExecutionException, InterruptedException {
        // 两个 CompletableFuture 组合执行，执行完后按顺序返回结果,观察到执行结果是5秒
        String res = CompletableFuture.supplyAsync(() -> {
            sleep(3000);
            return "结果1";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            sleep(5000);
            return "结果2";
        }), (res1, res2) -> {
            return res1 + "-" + res2;
        }).get();
        log.info(res);
    }

    @Test
    public void test12() throws ExecutionException, InterruptedException {
        // 两个 CompletableFuture 组合执行，执行完后按顺序返回结果,观察到执行结果是5秒
        String res = CompletableFuture.supplyAsync(() -> {
            sleep(3000);
            return "结果1";
        }).thenCombineAsync(CompletableFuture.supplyAsync(() -> {
            sleep(5000);
            return "结果2";
        }), (res1, res2) -> {
            return res1 + "-" + res2;
        }).get();
        log.info(res);
    }

    public void sleep(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            log.error("error", e);
            Thread.currentThread().interrupt();
        }
    }

    @Test
    public void test13() {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            sleep(3000);
            return "结果1";
        });
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            sleep(5000);
            return "结果2";
        });
        CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> {
            sleep(6000);
            return "结果3";
        });
        List<CompletableFuture<String>> completableFutures = Arrays.asList(cf1, cf2, cf3);
        //等待任意 CompletableFuture 执行完毕
        CompletableFuture.anyOf(cf1, cf2, cf3).thenAccept((r) -> {
            //List<String> rest = completableFutures.stream().map(CompletableFutureTest::getValue).collect(Collectors.toList());
            log.info(r.toString());
        }).join();
        log.info(String.valueOf(cf1));
        log.info(String.valueOf(cf2));
        log.info(String.valueOf(cf3));

    }


    @Test
    public void test14() throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            sleep(3000);
            return "结果1";
        });
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            sleep(5000);
            return "结果2";
        });
        CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> {
            sleep(6000);
            return "结果3";
        });
        //等待所有 CompletableFuture 执行完毕
        CompletableFuture.allOf(cf1, cf2, cf3).thenAccept((r) -> {
            System.out.println("执行完毕");
            System.out.println(cf1.join());
            System.out.println(cf2.join());
            System.out.println(cf3.join());
        }).get();
    }

    /**
     * 按执行顺讯返回结果
     */
    @Test
    public void test15() {
        List<String> res = executeAll(() -> {
            sleep(3000);
            log.info("执行1.。。。");
            return "结果1";
        }, () -> {
            sleep(4000);
            log.info("执行2.。。。");
            return "结果2";
        }, () -> {
            sleep(2000);
            log.info("执行3.。。。");
            return "结果3";
        });
        log.info(String.valueOf(res));
    }

    /**
     * 按返回顺序
     */
    @Test
    public void test16() {
        List<String> res = executeSequenceSuccessAll(() -> {
            sleep(1000);
            log.info("执行1.。。。");
            return "结果1";
        }, () -> {
            sleep(5000);
            log.info("执行2.。。。");
            return "结果2";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        }, () -> {
            sleep(3000);
            log.info("执行3.。。。");
            return "结果3";
        });
        log.info(String.valueOf(res));
    }

    public static <T> T getValue(CompletableFuture<T> future) {
        try {
          T t= future.get(10, TimeUnit.SECONDS);
            log.info("t...[{}]",t);
            return t;
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;
    }

    public static <T> List<T> executeAll(List<CompletableFuture<T>> futures) {
        return getValue(CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).thenApply(r -> {
            //这里的stream改成 parallelStream 则返回结果的顺序就变成了谁先执行完 返回谁
            return futures.stream().map(CompletableFutureTest::getValue).collect(Collectors.toList());
        }));
    }

    public static <T> List<T> sequenceSuccessFuture(List<CompletableFuture<T>> futures) {
        return futures.parallelStream().map(CompletableFuture::join).collect(Collectors.toList());
       /* return getValue(CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).thenApply(r -> {
            return futures.parallelStream().map(CompletableFutureTest::getValue).collect(Collectors.toList());
        }));*/
    }

    public static <T> List<T> executeAll(Supplier<T>... chains) {
        List<CompletableFuture<T>> cfs = Arrays.stream(chains).map(f -> {
            return CompletableFuture.supplyAsync(f);
        }).collect(Collectors.toList());
        return executeAll(cfs);
    }

    public static <T> List<T> executeSequenceSuccessAll(Supplier<T>... chains) {
        List<CompletableFuture<T>> cfs = Arrays.stream(chains).map(f -> {
            return CompletableFuture.supplyAsync(f);
        }).collect(Collectors.toList());
        return sequenceSuccessFuture(cfs);
    }
}
